package com.hsl.study.utils;


import lombok.extern.slf4j.Slf4j;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.TransactionException;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.interceptor.DefaultTransactionAttribute;

import java.util.ArrayList;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @Author: hsl
 * @CreateTime: 2025-02-15  11:42
 */
@Slf4j
public class MoreThreadTransactionUtils {
    public static boolean execute(PlatformTransactionManager platformTransactionManager, Runnable... taskList) {
        if (taskList == null || taskList.length == 0) {
            throw new IllegalArgumentException("taskList  is   null");
        }
        //任务数量
        int taskSize = taskList.length;
        //任务成功数量计数器
        AtomicInteger taskSuccessAccount = new AtomicInteger(0);

        ArrayList<Future<?>> taskFutureList = new ArrayList<>(taskSize);

        //循环屏障  用于多线程事务一起提交 || 一起回滚
        CyclicBarrier cyclicBarrier = new CyclicBarrier(taskSize);
        int i = 1;
        //定义一个线程池  线程池核心线程数大小和任务大小必须一致，因为里面的任务都必须同时去执行，否则会死锁
        ExecutorService executorService = Executors.newFixedThreadPool(taskSize);

        try {
            // 使用线程池执行循环处理任务  每个任务会交给线程池里面的线程进行执行
            for (Runnable runnable : taskList) {
                final int taskIndex = i;

                Future<?> future = executorService.submit(() -> {
                    TransactionStatus transactionStatus = null;

                    try {
                        // 使用spring编程式事务 开启事务
                        transactionStatus = platformTransactionManager.getTransaction(new DefaultTransactionAttribute());
                        // 执行任务
                        runnable.run();
                        // 成功数+ 1
                        taskSuccessAccount.incrementAndGet();
                        log.debug("runnable: {} 等待事务提交", taskIndex);
                    } catch (TransactionException e) {
                        log.error("runnable：{}，执行异常，异常原因：{}", taskIndex, e.getMessage());
                    } finally {
                        // 走到这里，会阻塞，直到当前线程池中所有的任务都执行到这个位置后，才会被唤醒，继续向下走
                        try {
                            cyclicBarrier.await();
                        } catch (Exception e) {
                            log.error("cyclicBarrier.await error:{}", e.getMessage(), e);
                        }
                    }

                    if (transactionStatus != null) {
                        //如果所有任务都成功 则一起提提交事务 但凡有失败 则回滚
                        if (taskSuccessAccount.get() == taskSize) {
                            //成功 事务提交
                            log.debug("runnable：{} 提交事务", taskIndex);
                            platformTransactionManager.commit(transactionStatus);
                        } else {
                            log.debug("runnable：{} 事务回滚", taskIndex);
                            platformTransactionManager.rollback(transactionStatus);
                        }
                    }
                });
                taskFutureList.add(future);
                i++;
            }

            for (Future<?> future : taskFutureList) {
                try {
                    future.get();
                } catch (Exception e) {
                    log.error("future.get error:{}", e.getMessage(), e);
                }
            }
        } finally {
            //关闭线程池
            executorService.shutdown();
        }
        // 如果所有任务都成功完成，则返回true；否则返回false
        return taskSuccessAccount.get() == taskSize;
    }
}